package com.shujia.streaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.ReceiverInputDStream

object Demo3Window {
  def main(args: Array[String]): Unit = {
    // 每隔半小时 统计最近一个小时内的热门微博
    // 每隔5s 统计最近10s的单词数量

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local[2]") // 接收数据需要占用一个线程
    conf.setAppName("Demo2UpdateStateByKey")
    val sc: SparkContext = new SparkContext(conf)
    // 创建StreamingContext的环境，设置处理时间间隔为5s
    // 相当于5s会封装成一个batch
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    // 使用有状态算子需要设置checkpoint地址
    ssc.checkpoint("spark/data/stream_checkpoint")

    /**
     * 使用nc工具模拟消息队列
     * 在linux中安装：yum install nc
     * nc -lk 8888
     *
     */
    // 通过socket连接到nc服务，得到DStream
    val words: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)

    words.flatMap(_.split(","))
      .map(word => (word, 1))
      // 滑动窗口
      // 如果滑动时间比窗口大小要小 => 滑动窗口
      // 如果滑动时间和窗口大小相等 => 滚动窗口
      // 一般滑动时间是小于窗口大小的 等于创建SparkStreamingContext指定的batch大小
      .reduceByKeyAndWindow(
        (x: Int, y: Int) => x + y // 传入一个聚合函数
        , (x: Int, y: Int) => x - y // 额外传入一个函数 对滑动窗口计算的一个优化 增对多余的数据怎么处理
        , Durations.seconds(10) // 窗口的大小为10s
        , Durations.seconds(5) // 每5s滑动一次
      )
      .print()




    // SparkStreaming程序需要手动启动
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
